package collector

import (
	"context"
	"fmt"
	"strconv"
	"strings"

	"github.com/go-redis/redis/v8"
	"gopkg.in/alecthomas/kingpin.v2"
	"regexp"
	"sync"
	"time"

	"github.com/go-kit/kit/log"
	"github.com/go-kit/kit/log/level"
	"github.com/prometheus/client_golang/prometheus"
)

// Metric name parts.
const (
	// Subsystem(s).
	exporter = "exporter"
)

var (
	versionRE = regexp.MustCompile(`^\d+\.\d+`)
)

// Tunable flags.
var (
	exporterLockTimeout = kingpin.Flag(
		"exporter.lock_wait_timeout",
		"Set a lock_wait_timeout (in seconds) on the connection to avoid long metadata locking.",
	).Default("2").Int()
)

// Metric descriptors.
var (
	scrapeDurationDesc = prometheus.NewDesc(
		prometheus.BuildFQName(namespace, exporter, "collector_duration_seconds"),
		"Collector time duration.",
		[]string{"collector"}, nil,
	)
)

// Verify if Exporter implements prometheus.Collector
var _ prometheus.Collector = (*Exporter)(nil)

// Exporter collects Redis metrics. It implements prometheus.Collector.
type Exporter struct {
	ctx      context.Context
	logger   log.Logger
	address  string
	passwd   string
	timeout  int
	scrapers []Scraper
	metrics  Metrics
}

// New returns a new Redis exporter for the provided url.
func New(ctx context.Context, address string, passwd string, metrics Metrics, scrapers []Scraper, logger log.Logger) *Exporter {
	timeout := *exporterLockTimeout

	return &Exporter{
		ctx:      ctx,
		logger:   logger,
		address:  address,
		passwd:   passwd,
		timeout:  timeout,
		scrapers: scrapers,
		metrics:  metrics,
	}
}

// Describe implements prometheus.Collector.
func (e *Exporter) Describe(ch chan<- *prometheus.Desc) {
	ch <- e.metrics.TotalScrapes.Desc()
	ch <- e.metrics.Error.Desc()
	e.metrics.ScrapeErrors.Describe(ch)
	ch <- e.metrics.RedisUp.Desc()
}

// Collect implements prometheus.Collector.
func (e *Exporter) Collect(ch chan<- prometheus.Metric) {
	e.scrape(e.ctx, ch)

	ch <- e.metrics.TotalScrapes
	ch <- e.metrics.Error
	e.metrics.ScrapeErrors.Collect(ch)
	ch <- e.metrics.RedisUp
}

func (e *Exporter) scrape(ctx context.Context, ch chan<- prometheus.Metric) {
	e.metrics.TotalScrapes.Inc()

	var err error

	scrapeTime := time.Now()

	// Create Redis Client
	addr := e.address
	password := e.passwd
	client := redis.NewClient(&redis.Options{
		Addr:     addr,
		Password: password,
	})
	defer client.Close()

	// Ping Redis
	startPing := time.Now()
	_, err = client.Ping(context.TODO()).Result()
	if err != nil {
		level.Error(e.logger).Log("msg", "Error pinging redis", "err", err)
		e.metrics.RedisUp.Set(0)
		e.metrics.Error.Set(1)
		return
	}

	e.metrics.RedisUp.Set(1)
	e.metrics.Error.Set(0)

	duration := time.Since(startPing)
	msg := fmt.Sprintf("Pinging redis duration %d ms.", duration.Milliseconds())
	level.Info(e.logger).Log("msg", msg)

	ch <- prometheus.MustNewConstMetric(scrapeDurationDesc, prometheus.GaugeValue, time.Since(scrapeTime).Seconds(), "connection")

	version := getRedisVersion(client, e.logger)
	var wg sync.WaitGroup
	defer wg.Wait()
	for _, scraper := range e.scrapers {
		if version < scraper.Version() {
			continue
		}

		wg.Add(1)
		go func(scraper Scraper) {
			defer wg.Done()
			label := "collect." + scraper.Name()
			scrapeTime := time.Now()
			if err := scraper.Scrape(ctx, client, ch, log.With(e.logger, "scraper", scraper.Name())); err != nil {
				level.Error(e.logger).Log("msg", "Error from scraper", "scraper", scraper.Name(), "err", err)
				e.metrics.ScrapeErrors.WithLabelValues(label).Inc()
				e.metrics.Error.Set(1)
			}
			ch <- prometheus.MustNewConstMetric(scrapeDurationDesc, prometheus.GaugeValue, time.Since(scrapeTime).Seconds(), label)
		}(scraper)
	}
}

// Metrics represents exporter metrics which values can be carried between http requests.
type Metrics struct {
	TotalScrapes prometheus.Counter
	ScrapeErrors *prometheus.CounterVec
	Error        prometheus.Gauge
	RedisUp      prometheus.Gauge
}

// NewMetrics creates new Metrics instance.
func NewMetrics() Metrics {
	subsystem := exporter
	return Metrics{
		TotalScrapes: prometheus.NewCounter(prometheus.CounterOpts{
			Namespace: namespace,
			Subsystem: subsystem,
			Name:      "scrapes_total",
			Help:      "Total number of times Redis was scraped for metrics.",
		}),
		ScrapeErrors: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: namespace,
			Subsystem: subsystem,
			Name:      "scrape_errors_total",
			Help:      "Total number of times an error occurred scraping a Redis.",
		}, []string{"collector"}),
		Error: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Subsystem: subsystem,
			Name:      "last_scrape_error",
			Help:      "Whether the last scrape of metrics from Redis resulted in an error (1 for error, 0 for success).",
		}),
		RedisUp: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "up",
			Help:      "Whether the Redis server is up.",
		}),
	}
}

// Version of redis
func getRedisVersion(client *redis.Client, logger log.Logger) float64 {
	//var versionStr string
	var versionNum float64

	reply, err := client.Info(context.TODO(), "server").Result()
	if err != nil {
		level.Error(logger).Log("msg", "Error getting info", "err", err)
	}

	data := make(map[string]string)

	// 处理响应字符串
	arr := strings.Split(reply, "\n")
	for _, field := range arr {
		if strings.HasPrefix(field, "#") {
			continue
		}
		kv := strings.Split(field, ":")
		if len(kv) > 1 {
			data[kv[0]] = kv[1]
		}
	}

	versionStr := data["redis_version"]

	// If we can't match/parse the version, set it some big value that matches all versions.
	versionNum, _ = strconv.ParseFloat(versionRE.FindString(versionStr), 64)
	if versionNum == 0 {
		level.Debug(logger).Log("msg", "Error parsing version string", "version", versionStr)
		versionNum = 999
	}

	return versionNum
}
